package com.yanggu.spark.core.serialize

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 从计算的角度看, RDD算子以内的代码都是在executor端执行, 其他的代码都是在driver端执行。
 * 在scala编程中经常会遇到算子内的代码引用到算子外的变量, 这样就形成了闭包的效果。
 * 如果算子外的数据无法进行序列化, 那就意味着driver端的数据无法传递给executor端, 造成计算错误
 * 因此在执行程序前, spark框架会对程序进行闭包检测。从源码中也可以看到这一点
 */
object Serialize_Fun01 {

  def main(args: Array[String]): Unit = {
    //1. 创建sparkConf对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Serialize_Fun01")

    //2. 创建sparkContext上下文环境对象
    val sc = new SparkContext(sparkConf)

    //3.创建一个RDD
    val rdd = sc.makeRDD(Array("hello world", "hello spark", "hive", "atguigu"))

    //3.1 创建Search对象
    val search = new Search("hello")

    //3.2 函数传递，打印：ERROR Task not serializable
    search.getMatch1(rdd).collect().foreach(println)

    //3.3 属性传递，打印：ERROR Task not serializable
    search.getMatch2(rdd).collect().foreach(println)

    //4. 关闭资源
    sc.stop()
  }

}

class Search(query:String)/* extends Serializable*/ {

  def isMatch(s: String): Boolean = {
    s.contains(query)
  }

  // 函数序列化案例
  def getMatch1 (rdd: RDD[String]): RDD[String] = {
    //rdd.filter(this.isMatch)
    rdd.filter(isMatch)
  }

  // 属性序列化案例
  def getMatch2(rdd: RDD[String]): RDD[String] = {
    //rdd.filter(x => x.contains(this.query))
    rdd.filter(x => x.contains(query))
    //val q = query
    //rdd.filter(x => x.contains(q))
  }
}
